package cn.tedu.process.flink;

import cn.tedu.process.param.ChargingProgressParam;
import cn.tedu.process.param.EMQXMetaData;
import com.alibaba.fastjson2.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.Properties;

@Configuration
public class FlinkRemoteProcessor {
    @PostConstruct
    public void init() throws Exception {
        // 定义远程 Flink 集群的配置
        String jobManagerHost = "192.168.150.129"; // 替换为你的 JobManager 主机名或 IP 地址
        int jobManagerPort = 6123; // 替换为你的 JobManager 端口，默认端口通常是 6123
        // 设置流执行环境
        // 创建指向远程 Flink 集群的执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
                jobManagerHost,
                jobManagerPort);
                //"/path/to/jarFile.jar"); // 如果 Flink 作业依赖外部的 jar 文件，需要在这里提供 第三个参数

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.150.129:9092"); // kafka链接broker
        properties.setProperty("group.id", "flink_consumer");

        // 创建kafka消费者
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                "chargingDataTopic", new SimpleStringSchema(), properties);

        // 添加资源来源到执行环境
        DataStream<String> stream = env.addSource(consumer);
        // 数据流处理：解析JSON并进行温度监控
        DataStream<ChargingProgressParam> processedStream = stream
                .map(new MapFunction<String, ChargingProgressParam>() {
                    @Override
                    public ChargingProgressParam map(String value) throws Exception {
                        EMQXMetaData emqxData = JSON.parseObject(value, EMQXMetaData.class);
                        if (emqxData.getPayload() != null) {
                            return JSON.parseObject(emqxData.getPayload(), ChargingProgressParam.class);
                        }
                        return null;
                    }
                })
                .filter(data -> data != null);

        // 路径1：温度超过45度时发出警告
        // 数据流处理：解析JSON并进行温度监控，应用滑动窗口
        processedStream
                .keyBy(data -> data.getGunId()) // 根据枪ID分组
                .window(SlidingProcessingTimeWindows.of(Time.minutes(2), Time.minutes(1))) // 2分钟的窗口，每1分钟滑动
                .apply(new WindowFunction<ChargingProgressParam, String, Integer, TimeWindow>() {
                    @Override
                    public void apply(Integer gunId, TimeWindow window, Iterable<ChargingProgressParam> values, Collector<String> out) {
                        double sumTemperature = 0;
                        int count = 0;
                        for (ChargingProgressParam data : values) {
                            sumTemperature += data.getTemperature();
                            count++;
                        }
                        double avgTemperature = sumTemperature / count;
                        if (avgTemperature > 45) {
                            // 构建警告信息
                            String alert = "警告：温度过高，当前平均温度：" + avgTemperature + "度，订单ID：" + values.iterator().next().getBillId() + "，枪ID：" + gunId;
                            out.collect(alert);
                            // 发送警告信息，例如通过消息系统
                            messageCharging(values.iterator().next(), false);
                        }
                    }
                })
                .print();

        // 路径2：温度超过48度时断开充电
        processedStream
                .filter(data -> data.getTemperature() > 48)
                .map(new MapFunction<ChargingProgressParam, String>() {
                    @Override
                    public String map(ChargingProgressParam data) throws Exception {
                        messageCharging(data,true);
                        return "充电已断开，订单ID：" + data.getBillId() + "，枪ID：" + data.getGunId();
                    }
                })
                .print();

        // 执行Flink作业
        env.execute("wendu");
    }
    // 该方法可以像rabbitmq或者/kafka 发送信息，最终由站内信推送
    private static void messageCharging(ChargingProgressParam data,boolean isDisconnect) {
        if(isDisconnect){
            //推送rabbitmq站内信断开通知
            //websocketMessage = new WebsocketMessage(chargingProgressParam.getUserId(), 1, "告警信息", "A系统检测到充电异常,设备已被锁定,请拔出充电枪更换设备");
            //rabbitTemplate.convertAndSend(RabbitMQInsideMessageConfig.INSIDE_MESSAGE_EXCHANGE, RabbitMQInsideMessageConfig.ALERT_KEY, websocketMessage);
            System.out.println("执行断电：订单ID " + data.getBillId() + "，枪ID " + data.getGunId());
        }else{
            //警告通知
            //websocketMessage = new WebsocketMessage(chargingProgressParam.getUserId(), 1, "告警信息", "B系统检测到充电异常,你可根据现场情况选择继续充电,或支付订单后更换其他设备");
            //rabbitTemplate.convertAndSend(RabbitMQInsideMessageConfig.INSIDE_MESSAGE_EXCHANGE, RabbitMQInsideMessageConfig.ALERT_KEY, websocketMessage);
            System.out.println("警告：温度过高，当前温度：" + data.getTemperature() + "度，订单ID：" + data.getBillId() + "，枪ID：" + data.getGunId());
        }
    }
}
